package com.atguigu.flink.sqlfunction;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/3/5
 *
 * /**
 *  * Created by Smexy on 2023/3/6
 *  *      Catalog: 是目录。
 *  *              目的是将不同数据源的同库同表的数据源分开。
 *  *
 *  *          举例： 从oracle读取数据作为动态表
 *  *                      test1.t1
 *  *                从mysql读取数据作为动态表
 *  *                      test1.t1
 *  *
 *  *         select * from oracle.test1.t1
 *  *         select * from mysql.test1.t1
 *  */

public class Demo6_Catalog
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = " create table t1 ( id string,ts bigint, vc int  )" +
            " with ( " +
            " 'connector' = 'filesystem' ,   " +
            " 'path' =  'data/cep.txt' ,   " +
            "  'format' = 'csv' " +
            "      )                 ";

        //执行建表
        tableEnvironment.executeSql(createTableSql);

        //获取当前的catalog
        System.out.println(tableEnvironment.getCurrentCatalog()); // default_catalog
        System.out.println(tableEnvironment.getCurrentDatabase()); //default_database

        tableEnvironment.sqlQuery("select * from default_catalog.default_database.t1").execute().print();

    }
}
